package com.chenzhiling.study.datasource.RemoteFile

import com.chenzhiling.study.datasource.FileSchema
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.sources.{BaseRelation, TableScan}
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.{Row, SQLContext}

/**
 * @Author: CHEN ZHI LING
 * @Date: 2021/8/26
 * @Description: 远程文件数据源
 */
class RemoteFileRelation(override val sqlContext:SQLContext,parameter: Map[String, String])
  extends BaseRelation with TableScan{


  private val IP: String = "ip"
  private val PORT: String = "port"
  private val USERNAME: String = "username"
  private val PASSWORD: String = "password"
  private val PATH: String = "path"

  override def schema: StructType = FileSchema.schema

  override def buildScan(): RDD[Row] = {
    val ip: String = parameter(IP)
    val port: Int = parameter(PORT).toInt
    val username: String = parameter(USERNAME)
    val password: String = parameter(PASSWORD)
    val path: String = parameter(PATH)
    val sc: SparkContext = sqlContext.sparkContext
    new RemoteFileRdd(sc,ip,port,username,password,path)
  }
}
